package org.zjt.weather.demo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * DESC
 *
 * @author
 * @create 2017-05-23 下午2:30
 **/
public class WeatherJobStart {
    public static void main(String[] args) {
        Configuration config = new Configuration();
        config.set("fs.defaultFS", "hdfs://master:9000");
        config.set("yarn.resourcemanager.hostname", "master");
        System.setProperty("HADOOP_USER_NAME", "root");    //设置用户的角色
//		config.set("mapred.jar", "C:\\Users\\Administrator\\Desktop\\wc.jar");
//		config.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", ",");
        try {
            FileSystem fs = FileSystem.get(config);

            Job job = Job.getInstance(config);
            job.setJarByClass(WeatherJobStart.class);

            job.setJobName("WeatherJobStart");

            job.setMapperClass(WeatherMapper.class);
            job.setReducerClass(WeatherReducer.class);
            job.setMapOutputKeyClass(WeatherKey.class);
            job.setMapOutputValueClass(DoubleWritable.class);

            job.setPartitionerClass(WeatherPartition.class);
            job.setSortComparatorClass(WeatherSort.class);
            job.setGroupingComparatorClass(WeatherGroup.class);   //决定分入同一个reduce对象中，元素传入reduce方法的values的合并。

            job.setNumReduceTasks(3);

            job.setInputFormatClass(KeyValueTextInputFormat.class);

            FileInputFormat.addInputPath(job, new Path("/usr/input/weather"));

            Path outpath = new Path("/usr/output/weather");
            if (fs.exists(outpath)) {
                fs.delete(outpath, true);
            }
            FileOutputFormat.setOutputPath(job, outpath);

            boolean f = job.waitForCompletion(true);
            if (f) {
                System.out.println("执行完成");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
